package apiTest;

import java.io.ByteArrayInputStream;
import java.io.IOException;
import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.Iterator;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.HConstants;
import org.apache.hadoop.hbase.client.HTable;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.ResultScanner;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.log4j.Level;
import org.apache.log4j.Logger;

import beans.WebPage;

public class ReadFromHbase {
	private static Logger hbaseLogger = Logger.getLogger("org.apache");
	private static Logger logger = Logger.getLogger(ReadFromHbase.class);

	/**
	 * 读取hbase中的rowkey和指定列
	 * 
	 * @throws IOException
	 */
	public static void getAllRows() throws IOException {
		
		WebPage webPage = new WebPage();
		
		hbaseLogger.setLevel(Level.ERROR);
		String strTBName = "enterprise_webpage";
		// 这是读取hbase-site.xml这个文件中的hbase
		Configuration conf = HBaseConfiguration.create();
		conf.setLong(HConstants.HBASE_CLIENT_SCANNER_TIMEOUT_PERIOD, 21600000);
		conf.setLong(HConstants.HBASE_CLIENT_SCANNER_CACHING, 100);
		// 如果读取的是本地的hbase 就把这句放开
		// conf.set("hbase.zookeeper.quorum", "localhost");
		@SuppressWarnings("resource")
		HTable table = new HTable(conf, Bytes.toBytes(strTBName)); // get table
		System.out.println("aaaaaaaaaaaaaaa");
		int pagenum = 0;
		String siteUrl = null;
		logger.info("-------------------new Scan-------------------");
		Scan scan = new Scan();
		ResultScanner resultScanner = table.getScanner(scan); // queryAll
		Iterator<Result> results = resultScanner.iterator();

		while (results.hasNext()) {
			try {
				++pagenum;
				logger.info("已经抓取到Hbase的第" + pagenum + "条数据");
				Result result = results.next();
				// 获得rowkey
				logger.info("===================================== rowKey===================================== ");
				byte[] row = result.getRow();
				ByteArrayInputStream rowBis = new ByteArrayInputStream(row);
				String rowCharset = CharSetUtil.getStreamCharSet(rowBis, "utf-8");
				String rowString = new String(row, rowCharset);
				logger.info(rowString);
				logger.info("===================================== rowKey===================================== ");

				// 获取title
				logger.info("===================================== title =====================================");
				byte[] titleByte = result.getValue(Bytes.toBytes("p"), Bytes.toBytes("t"));
				if (titleByte == null || titleByte.length == 0) {
					continue;
				}
				ByteArrayInputStream titleBis = new ByteArrayInputStream(titleByte);
				String titleCharset = CharSetUtil.getStreamCharSet(titleBis, "utf-8");
				String titleString = new String(titleByte, titleCharset);
				logger.info(titleString);
				logger.info("===================================== title =====================================");

				// 获取content
				logger.info("===================================== content =====================================");
				byte[] contentValue = result.getValue(Bytes.toBytes("p"), Bytes.toBytes("c"));
				if (contentValue == null || contentValue.length == 0) {
					continue;
				}
				ByteArrayInputStream bai = new ByteArrayInputStream(contentValue);
				String a = CharSetUtil.getStreamCharSet(bai, "utf-8");
				String content = new String(contentValue, a);
				logger.info(content);
				logger.info("===================================== content =====================================");

				// 获取fetchTime
				logger.info("===================================== fetchTime =====================================");
				byte[] timeByte = result.getValue(Bytes.toBytes("f"), Bytes.toBytes("ts"));
				if (timeByte == null || timeByte.length == 0) {
					continue;
				}

				long bytes2Long = bytes2Long(timeByte);
				SimpleDateFormat sfd = new SimpleDateFormat("yyyy-MM-dd");
				Date d = new Date(bytes2Long);
				String time = sfd.format(d);

				// String timeString = new String(timeByte, "utf-8");

				logger.info(time);
				logger.info("===================================== fetchTime =====================================");
				
				logger.info("---------------- 创建索引... ----------------");
				webPage.setId(rowString);
				webPage.setTitle(titleString);
				webPage.setContent(content);
				webPage.setTime(time);
				webPage.setUrl(rowString);
				
				CRUDTest.createIndex(webPage, "index", "webpage", rowString);
				
				
				logger.info("---------------- 创建索引完毕 ----------------");
//				break;

			} catch (Exception e) {
				logger.error(e.toString(), e);
				logger.error(siteUrl);
				continue;
			}
		}
		resultScanner.close();
		logger.info("-----------------close scanner--------------------");
		
		
	}

	public static long bytes2Long(byte[] byteNum) {
		long num = 0;
		for (int ix = 0; ix < 8; ++ix) {
			num <<= 8;
			num |= (byteNum[ix] & 0xff);
		}
		return num;
	}

	public static void main(String[] args) {
		try {
			ReadFromHbase.getAllRows();
		} catch (IOException e) {
			e.printStackTrace();
		}
	}
}
